package com.xwj.KafkaSpark

import com.xwj.myclass.MyConnection
import org.apache.spark.streaming.{Minutes, Seconds}
import org.apache.spark.streaming.kafka.KafkaUtils
/**
  * Created by thinkjoy on 2017/8/10.
  * 消费者
  */
object KafkaSparkConsumer {

  def main(args: Array[String]): Unit = {
    val TOPIC="letter"
    val zookeeper="192.168.0.78:2181,192.168.0.75:2181,192.168.0.76:2181"
    val stream = new MyConnection().StreamConnect()
    stream.checkpoint("checkpoint")
    val topicMap = TOPIC.split(",").map((_,1)).toMap
    val lines=KafkaUtils.createStream(stream,zookeeper,"1",topicMap).map(_._2)
    val words =lines.flatMap(_.split(" "))//对单词进行分割
    val wordCounts = words.map(x => (x, 1L))//统计出现的次数
      .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
    wordCounts.print()//输出结果
    stream.start()
    stream.awaitTermination()
  }
}
